package com.shujia.flink.core

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala._

object Demo8CheckPoint {
  def main(args: Array[String]): Unit = {


    //流处理环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * 1、开启checkpoint
     * checkpoint：将flink计算的状态持久化到分布式文件系统(hdfs)中,
     * 当任务因为各种原因执行失败可以基于checkpoint保存的数据恢复任务，保证状态不丢失
     *
     */

    // 每 100-0ms 开始一次 checkpoint
    env.enableCheckpointing(10000)

    // 高级选项：

    // 设置模式为精确一次 (这是默认值)
    //EXACTLY_ONCE: 数据处理的唯一一次
    //AT_LEAST_ONCE: 至少一次，可能会重复
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃， 如果计算的状态很大，checkpoint需要更多的时间
    env.getCheckpointConfig.setCheckpointTimeout(60000)

    // 允许两个连续的 checkpoint 错误
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)

    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

    // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
    env.getCheckpointConfig.setExternalizedCheckpointCleanup(
      ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    //指定checkpoint 将状态保存的位置（hdfs）
    env.getCheckpointConfig.setCheckpointStorage("hdfs://master:9000/flink/checkpoint")

    //状态后端：
    //HashMapStateBackend先包状态存储在taskManager的内存中，checkpoint时将状态持久化到hdfs
    //env.setStateBackend(new HashMapStateBackend())


    //EmbeddedRocksDBStateBackend: 先将状态保存再taskManager的磁盘上，checkpoint时再将状态持久化到hdfs
    env.setStateBackend(new EmbeddedRocksDBStateBackend())

    /**
     * HashMapStateBackend:先放内存，效率高，状态如果太大内存会放不下
     * EmbeddedRocksDBStateBackend： 先放磁盘，不受状态大小的限制，效率第
     */


    /**
     * 2、读取数据
     */
    //读取socket中的数据-- 无界流，  nc -lk 8888
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 3、处理数据
     */
    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))

    //转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))

    //安装单词进行分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)

    //对单词求和
    //sum 需要指定一个下标
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    //打印结果
    countDS.print()

    /**
     * 启动flink任务
     */

    env.execute()

    /**
     * 任务执行失败重启任务，指定checkpoint路径恢复任务
     * flink run -t yarn-session -Dyarn.application.id=application_1667291942113_0011 -c com.shujia.flink.core.Demo8CheckPoint -s hdfs://master:9000/flink/checkpoint/32efe9d9805c9d78b9aa7f2acefeae5d/chk-42 flink-1.0-SNAPSHOT.jar
     *
     */

  }

}
